package com.runssnail.demo.kafka.jms;

import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.serializer.StringDecoder;
import kafka.utils.VerifiableProperties;
import org.apache.commons.lang3.Validate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.jms.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Kafka TopicSubscriber
 * <p/>
 * Created by zhengwei on 2015/8/27.
 */
public class KafkaTopicSubscriber implements TopicSubscriber {

    private final static Logger log = LoggerFactory.getLogger(KafkaTopicSubscriber.class);

    /**
     * 默认的超时时间, 单位ms
     */
    public static final long DEFAULT_TIMEOUT = 300;

    /**
     * 默认线程数等于当前系统可用cpu核心数
     */
    public static final int DEFAULT_THREAD_COUNT = Runtime.getRuntime().availableProcessors();

    private ConsumerConnector connector;

    /**
     * 主题
     */
    private final Topic topic;

    /**
     * 超时时间，单位ms
     */
    private long timeout = DEFAULT_TIMEOUT;

    /**
     * 消息监听器
     */
    private List<MessageListener> messageListeners = null;

    /**
     * 接收到消息后调用MessageListener
     */
    private ExecutorService executor;

    /**
     * 线程数量
     */
    private int threadCount = DEFAULT_THREAD_COUNT;

    /**
     * 是否已经初始化
     */
    private volatile boolean initialized = false;

    /**
     * 创建KafkaTopicSubscriber
     *
     * @param topic 主题
     * @param connector kafka 消费connector
     */
    public KafkaTopicSubscriber(String topic, ConsumerConnector connector) {
        Validate.notNull(topic);
        Validate.notNull(connector);

        this.topic = new DefaultTopic("topic");
        this.connector = connector;
    }

    /**
     * 初始化方法
     */
    private synchronized void init() {

        if (initialized) {
            return ;
        }

        // 可用cpu核数
        int availableProcessors = Runtime.getRuntime().availableProcessors();

        Validate.notEmpty(messageListeners, "messageListeners is required");
        Validate.isTrue(this.threadCount > 0 && this.threadCount <= availableProcessors, "threadCount must > 0 && <= " + availableProcessors);

        initialized = true;

        if (this.executor == null) {
            this.executor = Executors.newFixedThreadPool(threadCount);
        }
    }

    @Override
    public Topic getTopic() throws JMSException {
        return this.topic;
    }

    /**
     * 获取该订阅者的 NoLocal 属性
     *
     * @return
     * @throws JMSException
     */
    @Override
    public boolean getNoLocal() throws JMSException {
        throw new UnsupportedOperationException();
    }

    @Override
    public String getMessageSelector() throws JMSException {
        throw new UnsupportedOperationException();
    }

    @Override
    public MessageListener getMessageListener() throws JMSException {
        throw new UnsupportedOperationException();
    }

    @Override
    public void setMessageListener(MessageListener messageListener) throws JMSException {
        throw new UnsupportedOperationException("use the method setMessageListeners");
    }

    @Override
    public Message receive() throws JMSException {
        return this.receive(timeout);
    }

    @Override
    public Message receive(long timeout) throws JMSException {

        if (!initialized) {
            init();
        }

        Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
        topicCountMap.put(this.topic.getTopicName(), this.threadCount);

        StringDecoder keyDecoder = new StringDecoder(new VerifiableProperties());
//        StringDecoder valueDecoder = new StringDecoder(new VerifiableProperties());

        MessageDecoder valueDecoder = new MessageDecoder();

        Map<String, List<KafkaStream<String, Object>>> consumerMap = this.connector.createMessageStreams(topicCountMap,
                keyDecoder,
                valueDecoder);

        final List<KafkaStream<String, Object>> streams = consumerMap.get(this.topic.getTopicName());

        for (final KafkaStream<String, Object> stream : streams) {
            this.executor.execute(new MessageProcessor(stream));
        }

        return null;
    }

    /**
     * 消息处理器
     */
    class MessageProcessor implements Runnable {

        KafkaStream<String, Object> stream;

        public MessageProcessor(KafkaStream<String, Object> stream) {
            this.stream = stream;
        }

        @Override
        public void run() {

            if (log.isInfoEnabled()) {
                log.info("MessageProcessor start, topic=" + topic);
            }

            try {
                ConsumerIterator<String, Object> it = stream.iterator();

                // 没有消息的话，这里会阻塞
                while (it.hasNext()) {
                    try {
                        final Message message = (Message) it.next().message();
                        if (log.isDebugEnabled()) {
                            log.debug("received message, message=" + message);
                        }

                        if (messageListeners != null && !messageListeners.isEmpty()) {
                            for (MessageListener listener : messageListeners) {
                                try {
                                    listener.onMessage(message);
                                } catch (Exception e) {
                                    log.error("listener on message error, listener=" + listener + ", message=" + message, e);
                                }
                            }
                        }

                    } catch (Exception e) {
                        log.error("process message error", e);
                    }
                }
            } catch (Exception e) {
                log.error("message processor error, topic=" + topic, e);
            }
        }
    }

    @Override
    public Message receiveNoWait() throws JMSException {
        return this.receive();
    }

    @Override
    public void close() throws JMSException {
        this.connector.shutdown();
        this.executor.shutdown();
    }

    public long getTimeout() {
        return timeout;
    }

    public void setTimeout(long timeout) {
        this.timeout = timeout;
    }

    public ExecutorService getExecutor() {
        return executor;
    }

    public void setExecutor(ExecutorService executor) {
        this.executor = executor;
    }

    public List<MessageListener> getMessageListeners() {
        return messageListeners;
    }

    public void setMessageListeners(List<MessageListener> messageListeners) {
        this.messageListeners = messageListeners;
    }

    /**
     * 添加 MessageListener
     *
     * @param messageListener 监听器
     */
    public synchronized void addMessageListener(MessageListener messageListener) {
        if (this.messageListeners == null) {
            this.messageListeners = new ArrayList<MessageListener>();
        }

        this.messageListeners.add(messageListener);
    }
}
